package com.ultrapower.mapper;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;

import com.sematext.hbase.wd.RowKeyDistributorByHashPrefix;
import com.ultrapower.bean.StringMetaDataBean;
import com.ultrapower.util.HBaseTables;

public class StringMetaDataMapper implements RowMapper<List<StringMetaDataBean>> {

	private RowKeyDistributorByHashPrefix rowKeyDistributorByHashPrefix;

	public StringMetaDataMapper(RowKeyDistributorByHashPrefix rowKeyDistributorByHashPrefix) {
		this.rowKeyDistributorByHashPrefix = rowKeyDistributorByHashPrefix;
	}

	private final static String STRING_METADATA_CF_STR_QUALI_STRING = Bytes.toString(HBaseTables.STRING_METADATA_CF_STR_QUALI_STRING);

	@Override
	public List<StringMetaDataBean> mapRow(Result result, int rowNum) throws Exception {
		if (result.isEmpty()) {
			return Collections.emptyList();
		}
		final byte[] rowKey = getOriginalKey(result.getRow());

		List<StringMetaDataBean> stringMetaDataList = new ArrayList<>();
		Cell[] rawCells = result.rawCells();
		for (Cell cell : rawCells) {
			StringMetaDataBean sqlMetaDataBo = new StringMetaDataBean();
			sqlMetaDataBo.readRowKey(rowKey);
			String stringValue = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(),
					cell.getQualifierLength());

			if (STRING_METADATA_CF_STR_QUALI_STRING.equals(stringValue)) {
				stringValue = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
			}

			sqlMetaDataBo.setStringValue(stringValue);
			stringMetaDataList.add(sqlMetaDataBo);
		}
		return stringMetaDataList;
	}

	private byte[] getOriginalKey(byte[] rowKey) {
		return rowKeyDistributorByHashPrefix.getOriginalKey(rowKey);
	}
}
